package net.bwie.realtime.jtp.dwd.log.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class AppLogSplitProcessFunction extends ProcessFunction<String, String> {
    private final OutputTag<String> errorTag;
    private final OutputTag<String> startTag;
    private final OutputTag<String> displayTag;
    private final OutputTag<String> actionTag;


    public AppLogSplitProcessFunction(OutputTag<String> errorTag, OutputTag<String> startTag,
                                      OutputTag<String> displayTag, OutputTag<String> actionTag) {
        this.errorTag = errorTag;
        this.startTag = startTag;
        this.displayTag = displayTag;
        this.actionTag = actionTag;
    }

    @Override
    public void processElement(String s, ProcessFunction<String, String>.Context context,
                               Collector<String> collector) throws Exception {
        // 1.解析 json数据
        JSONObject jsonObject = JSON.parseObject(s);

        // 2. 错误日志
        if (null != jsonObject.get("err")) {
            // 侧边输出
            context.output(errorTag, s);
            // 错误数据单独存储
            jsonObject.remove("err");
        }

        // 3.启动日志
        if (null != jsonObject.get("start")) {
            //侧边流输出
            context.output(startTag, jsonObject.toJSONString());
        }
//       4. 页面日志
        else {
            // 曝光和动作数据
            JSONArray displaysjsonarray = jsonObject.getJSONArray("displays");
            JSONArray actioonsjsonarray = jsonObject.getJSONArray("actioons");

            //删除曝光和动作数据
            jsonObject.remove("displays");
            jsonObject.remove("actioons");

            // 输出页面日志
            collector.collect(jsonObject.toJSONString());

            // 5.曝光日志
            if (null != displaysjsonarray && !displaysjsonarray.isEmpty()) {

                for (int i = 0; i < displaysjsonarray.size(); i++) {
                    JSONObject displayjsonObject = displaysjsonarray.getJSONObject(i);
                    //添加字段值
                    jsonObject.put("display", displayjsonObject);
                    // 输出
                    context.output(displayTag, jsonObject.toJSONString());
                }
            }
            // 将上述添加曝光数据字段值删除
            jsonObject.remove("display");

            //6.动作日志
            if (null != actioonsjsonarray && !actioonsjsonarray.isEmpty()) {
                for (int i = 0; i < actioonsjsonarray.size(); i++) {
                    JSONObject actionjsonObject = actioonsjsonarray.getJSONObject(i);
                    //添加字段值
                    jsonObject.put("action", actionjsonObject);
                    // 输出
                    context.output(actionTag, jsonObject.toJSONString());
                }
            }

        }


    }
}
